from itertools import groupby

from pyspark.sql.functions import *
from pyspark.sql.session import SparkSession

spark = SparkSession.builder.master('local').appName('demo2_dsl').getOrCreate()

students_df = spark.read.format('json').load('../../data/students.json')

# 打印表的结构
students_df.printSchema()

# show:默认查看前20行数据
# students_df.show()
# 当设置为 False 时，会完整显示每一列的内容，不会出现省略号（...）。
# 当设置为 True 时，如果字段内容过长，则会进行截断显示。
# students_df.show(truncate=False)

# 2,where:过滤数据

# 使用sql表达式过滤数据

# students_df.where('age > 18').show()
# students_df.where('gender = "男"').show()
# students_df.where('age > 18 and gender = "男"').show()

# 使用列表表达式
# students_df.where(students_df.age == 21).show()

# col('age'):获取列

# students_df.where(col('age') == 21).show()
# students_df.where(col('age').isNotNull()).show()

# select
# students_df.select('id', 'name').show()
# students_df.select('id', (col('age') + 1).alias('age')).show()
# students_df.select('id', substring('clazz', 1, 2).alias('clazz_type')).show()

# selectExpr:使用sql表达式的方法
# students_df.selectExpr('id', 'substring(clazz,1,2) as clazz_type').show()

# group by分组聚合
# (
#     students_df
#     .groupby('clazz')
#     .agg(
#         count('clazz').alias('num'),
#         round(avg('age'), 2).alias('avg_age'),
#         max('age').alias('max_age')
#     )
#     .show()
# )

# 聚合之后进行过滤，相当于having
# students_df \
#     .groupby('clazz') \
#     .agg(count('clazz').alias('num')) \
#     .where(col('num') > 80) \
#     .show()
# order by
# students_df \
#     .groupby('clazz') \
#     .agg(count('clazz').alias('num')) \
#     .orderBy(col('num').desc()) \
#     .show()


#limit:返回DF
# students_df.limit(10).show()

#head:返回列表，将DF的数据拉去到Driver端
# print(students_df.head(10))

#join
#读取分数表
score_df = (spark.read.format('csv').schema('sid string, cid string, score double').option('sep', ',').load(
    '../../data/score.txt'))

#1,关联字段名一致
# students_df.join(score_df,'id',how='left').show()

#2,关联字段名不一致
# students_df.join(score_df,col('id')==col('sid'),how='inner').show()

#统计每个班级总督的平均分
students_df\
    .join(score_df,col('id') == col('sid'),how='inner')\
    .groupby('id','clazz')\
    .agg(sum('score').alias('sum_score'))\
    .groupby('clazz')\
    .agg(round(avg('sum_score'),2).alias('avg_score'))\
    .show()